/**
 * Created by lock
 * Date: 2019-08-12
 * Time: 23:36
 */
package connect

import (
	"context"
	"fmt"
	"github.com/rcrowley/go-metrics"
	"github.com/smallnest/rpcx/client"
	"github.com/smallnest/rpcx/server"
	"github.com/smallnest/rpcx/serverplugin"
	"go.uber.org/zap"
	"gochat/config"
	"gochat/proto"
	"gochat/tools"
	"strings"
	"sync"
	"time"
)

var logicRpcClient client.XClient
var once sync.Once

type RpcConnect struct {
}

func (c *Connect) InitLogicRpcClient() (err error) {
	once.Do(func() {
		d := client.NewEtcdV3Discovery(
			config.Conf.Common.CommonEtcd.BasePath,
			config.Conf.Common.CommonEtcd.ServerPathLogic,
			[]string{config.Conf.Common.CommonEtcd.Host},
			nil,
		)
		logicRpcClient = client.NewXClient(config.Conf.Common.CommonEtcd.ServerPathLogic,
											client.Failtry,
											client.RandomSelect,
											d,
											client.DefaultOption)
	})
	if logicRpcClient == nil {
		zap.L().Error("get rpc client nil")
		return
	}
	return
}

func (rpc *RpcConnect) Connect(connReq *proto.ConnectRequest) (uid int, err error) {
	reply := &proto.ConnectReply{}
	err = logicRpcClient.Call(context.Background(), "Connect", connReq, reply)
	if err != nil {
		zap.L().Panic("failed to call: ",zap.Error(err))
	}
	uid = reply.UserId
	zap.L().Info("connect logic userId :", zap.String("UserId",string(reply.UserId)))
	return
}

func (rpc *RpcConnect) DisConnect(disConnReq *proto.DisConnectRequest) (err error) {
	reply := &proto.DisConnectReply{}
	if err = logicRpcClient.Call(context.Background(), "DisConnect", disConnReq, reply); err != nil {
		zap.L().Fatal("failed to call: %v", zap.Error(err))
	}
	return
}

func (c *Connect) InitConnectWebsocketRpcServer() (err error) {
	var network, addr string
	connectRpcAddress := strings.Split(config.Conf.Connect.ConnectRpcAddressWebSockts.Address, ",")
	for _, bind := range connectRpcAddress {
		if network, addr, err = tools.ParseNetwork(bind); err != nil {
			zap.L().Panic("InitConnectWebsocketRpcServer ParseNetwork error : %s", zap.Error(err))
		}
		zap.L().Info("Connect start run at-->",
				zap.String("network",network),
				zap.String("addr",addr))
		go c.createConnectWebsocktsRpcServer(network, addr)
	}
	return
}

func (c *Connect) InitConnectTcpRcpServer() (err error) {
	var network, addr string
	connectRpcAddress := strings.Split(config.Conf.Connect.ConnectRpcAddressTcp.Address, ",")
	for _, bind := range connectRpcAddress {
		if network, addr, err = tools.ParseNetwork(bind); err != nil {
			zap.L().Panic("InitConnectTcpRcpServer ParseNetwork error : %s", zap.Error(err))
		}
		zap.L().Info("Connect start run at-->",
			zap.String("network",network),
			zap.String("addr",addr))
		go c.createConnectTcpRpcServer(network, addr)
	}
	return
}

type RpcConnectPush struct {
}

func (rpc *RpcConnectPush) PushSingleMsg(ctx context.Context, pushMsgReq *proto.PushMsgRequest, successReply *proto.SuccessReply) (err error) {
	var (
		bucket  *Bucket
		channel *Channel
	)
	zap.L().Info("rpc PushMsg : ",zap.Any("",pushMsgReq))
	if pushMsgReq == nil {
		zap.L().Error("rpc PushSingleMsg() args:",zap.Any("",pushMsgReq) )
		return
	}
	bucket = DefaultServer.Bucket(pushMsgReq.UserId)
	if channel = bucket.Channel(pushMsgReq.UserId); channel != nil {
		err = channel.Push(&pushMsgReq.Msg)
		zap.L().Info("DefaultServer Channel err nil ,args: ", zap.Any("pushMsgReq",pushMsgReq))
		return
	}
	successReply.Code = config.SuccessReplyCode
	successReply.Msg = config.SuccessReplyMsg
	zap.L().Info("successReply: ", zap.Any("",successReply))
	return
}

func (rpc *RpcConnectPush) PushRoomMsg(ctx context.Context, pushRoomMsgReq *proto.PushRoomMsgRequest, successReply *proto.SuccessReply) (err error) {
	successReply.Code = config.SuccessReplyCode
	successReply.Msg = config.SuccessReplyMsg

	zap.L().Info("PushRoomMsg msg %+v",zap.Any("",pushRoomMsgReq))
	for _, bucket := range DefaultServer.Buckets {
		bucket.BroadcastRoom(pushRoomMsgReq)
	}
	return
}

func (rpc *RpcConnectPush) PushRoomCount(ctx context.Context, pushRoomMsgReq *proto.PushRoomMsgRequest, successReply *proto.SuccessReply) (err error) {
	successReply.Code = config.SuccessReplyCode
	successReply.Msg = config.SuccessReplyMsg

	zap.L().Info("PushRoomCount msg %v",zap.Any("",pushRoomMsgReq))
	for _, bucket := range DefaultServer.Buckets {
		bucket.BroadcastRoom(pushRoomMsgReq)
	}

	return
}

func (rpc *RpcConnectPush) PushRoomInfo(ctx context.Context, pushRoomMsgReq *proto.PushRoomMsgRequest, successReply *proto.SuccessReply) (err error) {
	successReply.Code = config.SuccessReplyCode
	successReply.Msg = config.SuccessReplyMsg

	zap.L().Info("connect,PushRoomInfo msg %+v",zap.Any("",pushRoomMsgReq))
	for _, bucket := range DefaultServer.Buckets {
		bucket.BroadcastRoom(pushRoomMsgReq)
	}
	return
}

func (c *Connect) createConnectWebsocktsRpcServer(network string, addr string) {
	s := server.NewServer()
	addRegistryPlugin(s, network, addr)
	//config.Conf.Connect.ConnectTcp.ServerId
	s.RegisterName(config.Conf.Common.CommonEtcd.ServerPathConnect,
		new(RpcConnectPush),
		fmt.Sprintf("%s", config.Conf.Connect.ConnectWebsocket.ServerId))
	s.RegisterOnShutdown(func(s *server.Server) {
		s.UnregisterAll()
	})
	s.Serve(network, addr)
}

func (c *Connect) createConnectTcpRpcServer(network string, addr string) {
	s := server.NewServer()
	addRegistryPlugin(s, network, addr)
	s.RegisterName(config.Conf.Common.CommonEtcd.ServerPathConnect,
		new(RpcConnectPush),
		fmt.Sprintf("%s", config.Conf.Connect.ConnectTcp.ServerId))
	s.RegisterOnShutdown(func(s *server.Server) {
		s.UnregisterAll()
	})
	s.Serve(network, addr)
}

func addRegistryPlugin(s *server.Server, network string, addr string) {

	r := &serverplugin.EtcdV3RegisterPlugin{
		ServiceAddress: network + "@" + addr,							//本机的监听地址
		EtcdServers:    []string{config.Conf.Common.CommonEtcd.Host},	//etcd集群的地址
		BasePath:       config.Conf.Common.CommonEtcd.BasePath,			//服务前缀。 如果有多个项目同时使用zookeeper，避免命名冲突，可以设置这个参数，为当前的服务设置命名空间
		Metrics:        metrics.NewRegistry(),							//用来更新服务的TPS
		UpdateInterval: time.Minute,			//服务的刷新间隔， 如果在一定间隔内(当前设为2 * UpdateInterval)没有刷新,服务就会从etcd中删除
	}
	err := r.Start()
	if err != nil {
		zap.L().Fatal("",zap.Error(err))
	}
	s.Plugins.Add(r)
}
